testcontainersでPythonからMySQLへ接続してテストしてみた

アッキー
2024.12.20
こんにちは。製造ビジネステクノロジー部のakkyです。
PythonでMySQLなどのDBへの接続を自動テストしたい場合はどうすればよいでしょうか？

データ自体が大事な場面であればモックを使うのも手ですが、実際のDBに接続してのテストも行いたいことがあると思います。
DockerでMySQLのコンテナを起動して接続するのが簡単そうですが、テストのたびにコンテナを起動するのは少々面倒ですね。

そんな手間を削減できるライブラリとして、testcontainersがあります。
testcontainersは、モックではなく実際のコンテナを起動して指定したプログラムへ簡単に接続するまでのサポートをしてくれるライブラリです。
コミュニティ開発のため言語ごとに対応しているコンテナが異なっており、Pythonでは使用できるソフトウェアの種類が限られている印象がありますが、それでも有名なソフトウェアは対応している印象です。
https://testcontainers-python.readthedocs.io/en/latest/

https://github.com/testcontainers/testcontainers-python
今回はtestcontainersを使ってPythonでMySQLへ接続するコードをテストしてみたので紹介します。
 使用バージョンWindows 11 Pro
Python 3.12.7
testcontainers 4.9.0
Rancher Desktop 1.16.0
 インストール次のコマンドでインストールします
pip install testcontainers[mysql]
なお、testcontainers-mysqlはすでにメンテナンスされていないため、使わないでください。
 対象コードMySQLに接続し、値を読み書きするだけの簡単なコードです。

testcontainersのドキュメントではSQLAlchemyを使用する方法を紹介しているので、O/Rマッパーは使用せず、PyMySQLでべた書きしてみました。
import pymysql
import pymysql.cursors
import typing

def connect_db(host:str, user:str, password:str, database:str, port=3306):
    connection = pymysql.connect(host=host,user=user,password=password, database=database,port=port)
    return connection

def create_table(connection:pymysql.Connection):
    cursor:pymysql.cursors.Cursor = connection.cursor()
    cursor.execute("""CREATE TABLE greeting_table (
                   id INT NOT NULL PRIMARY KEY AUTO_INCREMENT, 
                   greeting_time VARCHAR(10), 
                   greeting_contents VARCHAR(32)
                   )""")
    connection.commit()
    cursor.close()

def write_data(connection:pymysql.Connection, greeting_time:str, greeting_contents:str):
    cursor:pymysql.cursors.Cursor = connection.cursor()
    cursor.execute("INSERT INTO greeting_table (greeting_time, greeting_contents) VALUES(%s,%s)",
                   (greeting_time, greeting_contents))
    connection.commit()
    cursor.close()

def read_data(connection:pymysql.Connection, greeting_time:str) -> tuple[str]:
    cursor:pymysql.cursors.Cursor = connection.cursor()
    cursor.execute("SELECT greeting_contents FROM greeting_table WHERE greeting_time=%s", 
                   (greeting_time,))
    result = typing.cast(tuple[str], cursor.fetchone())
    cursor.close()
    return result

def close_db(connection:pymysql.Connection):
    connection.close()
 testcontainersを使用したテストコードテストフレームワークにはunittestを使用しました。
 解説testcontainersによるMySQLの準備はsetUpClassに含まれています。
コンテナ自体はMySqlContainer(image="mysql:8")だけで準備できて、start()すると起動し、接続できるまで待機してくれます。
重要なのは接続先です。

ソースコードを見るのが早いですが、ユーザー名もパスワードもtestです。

ホスト名はget_container_host_ip()で取得できます。
問題はポート番号です。portで取得したポート番号に接続できずだいぶ悩んだのですが、これはコンテナ内部で使用されるポート番号で、これをget_exposed_port(cls.container.port)として外部に公開されるポート番号に変換する必要がありました。
import unittest
from testcontainers.mysql import MySqlContainer

import db_lib

class TestDBLib(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.container = MySqlContainer(image="mysql:8")
        cls.container.start()
        cls.connection = db_lib.connect_db(
            host=cls.container.get_container_host_ip(),
            port=int(cls.container.get_exposed_port(cls.container.port)),
            user="test",
            password="test",
            database="test")
        db_lib.create_table(cls.connection)

    @classmethod
    def tearDownClass(cls) -> None:
        db_lib.close_db(cls.connection)
        cls.container.stop()

    def test_read_write_success(self):
        """DB読み書き（正常系）"""
        db_lib.write_data(self.connection, "morning", "good morning")
        result = db_lib.read_data(self.connection, "morning")
        self.assertEqual(result[0], "good morning")

    def test_read_not_inserted_value(self):
        """DB読み出し（異常系）"""
        result = db_lib.read_data(self.connection, "night")
        self.assertIsNone(result)

if __name__ == '__main__':
    unittest.main()
 テストの実行unittestを利用したコードとして通常の手順で実行できます。
> python -m unittest
Pulling image testcontainers/ryuk:0.8.1
Container started: fe6e7e6ab023
Waiting for container <Container: fe6e7e6ab023> with image testcontainers/ryuk:0.8.1 to be ready ...
Pulling image mysql:8
Container started: 9be16584d388
Waiting for container <Container: 9be16584d388> with image mysql:8 to be ready ...
..
----------------------------------------------------------------------
Ran 2 tests in 23.801s

OK
以上